package com.orange.project.oneSay.service.impl;

import com.fasterxml.jackson.core.type.TypeReference;
import com.google.common.collect.Lists;
import com.orange.bean.core.constant.CacheConstants;
import com.orange.bean.core.enums.OneSayCategory;
import com.orange.bean.oneSay.dto.OneSayDto;
import com.orange.bean.oneSay.po.OneSayPo;
import com.orange.bean.oneSay.vo.OneSayReq;
import com.orange.bean.oneSay.vo.OneSayRes;
import com.orange.common.base.service.impl.BaseServiceImpl;
import com.orange.common.enums.ResultCodeEnum;
import com.orange.common.exception.BusinessException;
import com.orange.common.utils.ConvertUtils;
import com.orange.common.utils.HashUtils;
import com.orange.common.utils.JsonReader;
import com.orange.project.oneSay.mapper.OneSayMapper;
import com.orange.project.oneSay.service.OneSayService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.support.TransactionTemplate;

import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * TODO
 *
 * @Author YWF
 * @Date 2025/9/8 17:28
 */
@Service
public class OneSayServiceImpl extends BaseServiceImpl<OneSayMapper, OneSayPo> implements OneSayService {


    @Autowired
    private OneSayMapper oneSayMapper;

    @Autowired
    private TransactionTemplate transactionTemplate;

    /**
     * 初始化一言数据
     *
     * @return
     */
    @Override
    @Async("asyncServiceExecutor")
    public void getOneSayList(String folderPath) {
        long startTime = System.currentTimeMillis();
        try {
            // 1. 路径处理
            folderPath = Paths.get(folderPath).toString() + java.io.File.separator;

            // 2. 从文件加载数据到List
            List<OneSayDto> allHitokotos = readAllJsonFiles(folderPath);
            // 3. 使用并行流处理数据转换,并去重
            List<OneSayPo> allOneSayList = allHitokotos.parallelStream()
                    .collect(Collectors.toConcurrentMap(
                            OneSayDto::getHitokoto,
                            this::convertToPo,
                            (oldVal, newVal) -> oldVal
                    ))
                    .values().stream()
                    .collect(Collectors.toList());

            int totalCount = allOneSayList.size();
            logger.info("开始处理 {} 条数据", totalCount);

            // 4. 分批次并行入库（每批1000条）
            this.batchInsertToDatabase(allOneSayList, 1000);

            // 5. 批量缓存到Redis（分批次）
            this.batchCacheToRedis(allOneSayList, 1000);

            long endTime = System.currentTimeMillis();
            logger.info("初始化完成，共处理数据：{}条，耗时：{}ms", totalCount, (endTime - startTime));

        } catch (Exception e) {
            logger.error("数据处理过程中发生异常", e);
        }
    }

    /**
     * 批量插入数据库
     */
    private void batchInsertToDatabase(List<OneSayPo> dataList, int partSize) {
        AtomicInteger batchCount = new AtomicInteger(0);
        // 分片处理数据
        Lists.partition(dataList, partSize).forEach(batch -> {
            int currentBatch = batchCount.incrementAndGet();
            // 使用事务确保批次操作的原子性
            transactionTemplate.execute(status -> {
                try {
                    oneSayMapper.batchOneSay(batch);
                    if (currentBatch % 10 == 0) { // 每10批记录一次
                        logger.info("已入库 {} 批数据，当前批大小: {}", currentBatch, batch.size());
                    }
                    return null;
                } catch (Exception e) {
                    logger.error("第 {} 批数据入库失败", currentBatch, e);
                    status.setRollbackOnly(); // 回滚当前批次
                    throw new RuntimeException("数据库插入失败", e);
                }
            });
        });
    }


    /**
     * 对象转换
     *
     * @param dto
     * @return
     */
    private OneSayPo convertToPo(OneSayDto dto) {
        OneSayPo po = ConvertUtils.cloneObject(dto, OneSayPo.class);
        po.setUid(HashUtils.md5(dto.getHitokoto()));
        po.setTypeN(OneSayCategory.getByKey(po.getType()).getName());
        po.setContent(dto.getHitokoto());
        po.setFromSource(dto.getFrom());
        po.setFromWho(dto.getFrom_who());
        po.setCommitFrom(dto.getCommit_from());
        po.setCommitWho("一言语料库");
        return po;
    }

    /**
     * 读取目录下所有 JSON 文件并合并数据
     */
    public static List<OneSayDto> readAllJsonFiles(String folderPath) {
        List<OneSayDto> result = new ArrayList<>();
        // 遍历 a.json 到 i.json
        for (char c = 'a'; c <= 'i'; c++) {
            String fileName = c + ".json";
            List<OneSayDto> data = null;
            try {
                data = JsonReader.readJsonFile(folderPath + fileName, new TypeReference<List<OneSayDto>>() {
                });
                if (data != null) {
                    result.addAll(data);
                }
            } catch (Exception exception) {
                logger.error("OneSayServiceImpl-->readAllJsonFiles数据解析失败:{} ", exception.getMessage());
            }
        }

        return result;
    }

    /**
     * 批量缓存到Redis
     */
    private void batchCacheToRedis(List<OneSayPo> allOneSayList, int partSize) {
        AtomicInteger processedCount = new AtomicInteger(0);
        Lists.partition(allOneSayList, partSize).forEach(batch -> {
            redisCache.sSet(CacheConstants.ONE_SAY_LIST, batch);
            processedCount.addAndGet(batch.size());
            if (processedCount.get() % partSize == 0) {
                logger.info("已缓存 {}/{}", processedCount.get(), allOneSayList.size());
            }
        });
        logger.info("Redis缓存完成，共缓存 {} 条数据", allOneSayList.size());
    }

    @Override
    public OneSayRes getRandomOneSay() {
        OneSayRes randomOneSay = redisCache.getRandomFromSet(CacheConstants.ONE_SAY_LIST, OneSayRes.class);
        logger.info("随机获取一条数据：{}", randomOneSay);
        return randomOneSay;
    }

    @Override
    public OneSayRes saveCommitOneSay(OneSayReq oneSayReq) {
        OneSayPo oneSayPo = ConvertUtils.cloneObject(oneSayReq, OneSayPo.class);
        oneSayPo.setUid(HashUtils.md5(oneSayReq.getContent()));
        oneSayPo.setLength(oneSayReq.getContent().length());
        int counts = oneSayMapper.insert(oneSayPo);
        if (counts > 0) {
            oneSayPo.setTypeN(OneSayCategory.getByKey(oneSayReq.getType()).getName());
            redisCache.sSet(CacheConstants.ONE_SAY_LIST, oneSayPo);
            OneSayRes oneSayRes = ConvertUtils.cloneObject(oneSayPo, OneSayRes.class);
            return oneSayRes;
        }
        throw new BusinessException(ResultCodeEnum.SAVE_ONE_SAY_ERROR);
    }

    @Override
    @Async("asyncServiceExecutor")
    public void batchCommitOneSay(List<OneSayReq> oneSayListReq) {
        long startTime = System.currentTimeMillis();

        try {
            // 1. 使用并行流处理数据转换,并去重
            List<OneSayPo> allOneSayList = oneSayListReq.parallelStream()
                    .collect(Collectors.toConcurrentMap(
                            OneSayReq::getContent,
                            this::convertToPo,
                            (firstOneSayPo, lastOneSayPo) -> firstOneSayPo
                    ))
                    .values().stream()
                    .collect(Collectors.toList());

            int totalCount = allOneSayList.size();
            logger.info("开始处理 {} 条数据", totalCount);

            // 2. 分批次并行入库（每批1000条）
            this.batchInsertToDatabase(allOneSayList, 10);

            // 3. 批量缓存到Redis（分批次）
            this.batchCacheToRedis(allOneSayList, 10);

            long endTime = System.currentTimeMillis();
            logger.info("初始化完成，共处理数据：{}条，耗时：{}ms", totalCount, (endTime - startTime));
        } catch (Exception e) {
            logger.error("数据处理过程中发生异常", e);
        }
    }

    /**
     * 对象转换
     *
     * @param o
     * @return
     */
    private OneSayPo convertToPo(OneSayReq o) {
        OneSayPo oneSayPo = ConvertUtils.cloneObject(o, OneSayPo.class);
        oneSayPo.setUid(HashUtils.md5(o.getContent()));
        oneSayPo.setTypeN(OneSayCategory.getByKey(oneSayPo.getType()).getName());
        oneSayPo.setLength(o.getContent().length());
        return oneSayPo;
    }
}
